package cn.doitedu.day05.demo

import java.sql.{DriverManager, PreparedStatement, ResultSet}

import cn.doitedu.day01.utils.SparkUtil
import org.apache.spark.rdd.{JdbcRDD, RDD}

/**
 * @Date 22.4.1
 * @Created by HANGGE
 * @Description
 */
object Demo02_订单02 {
  /**
   *       日志数据  (文件中)
   * 订单ID  商品类别  商品金额  商品ID
   * 1001,办公用品,900,1
   * 1002,生活用品,88,2
   * 1003,服装,699,3
   * 1004,办公用品,800,4
   * 1005,补品,9.9,6
   * 1006,服装,899,5
   * 1007,补品,99.88,7
   *
   *  商品信息 在数据库中
   * create table tb_goods(
   * pid int ,
   * name varchar(50),
   * describe  varchar(50)
   * ) ;
   * insert into tb_goods values
   * 1,"cherry机械键盘","just so so"),
   * 2,"雕牌卫生纸","厚而不硬"),
   * 3,"梦之蓝男装-格子衬衫","帅无边"),
   * 4,"奥妙打印机","你猜不到的结果"),
   * 5,"劲霸男装-短裤",""),
   * 6,"九全大补丸","就差那么一点点"),
   * 7,"多易牌黑枸杞","学大数, 不费劲") ;
   *
   * 1) 每种类别下的商品总金额
   * 2) 关联MySQL库中的商品名和描述信息 , 将结果存储到HDFS上
   *            - map算子
   *            - mapPartitions算子
   *            - MySQL数据构建RDD join
   *            - 广播变量
   */

  def main(args: Array[String]): Unit = {

    val sc = SparkUtil.getSc
    // 加载订单数据
    val data = sc.textFile("data/order.txt")
    // 处理每行订单
    val ordersRDD = data.map(line => {
      val arr = line.split(",")
      //1005,补品,9.9,6
      (arr(3).toInt , line)
    })

    ordersRDD.mapPartitionsWithIndex((p,iters)=>{
      iters.map(tp=>tp + "---"+p)
    }).foreach(println)

    val  goodsRDD = new JdbcRDD[(Int ,String)](
        sc ,
      ()=>DriverManager.getConnection("jdbc:mysql://localhost:3306/doit30?characterEncoding=UTF-8", "root", "root") ,
      "select * from tb_goods where ? <= pid and pid <=?" ,
      1 ,
      7 ,
      2 ,
      (rs:ResultSet)=>{
        val pid = rs.getInt(1)
        val name = rs.getString(2)
        val des = rs.getString(3)
        (pid , name+","+des)
      }
    )



    println(goodsRDD.getNumPartitions)

    goodsRDD.mapPartitionsWithIndex((p , iters)=>{
      iters.map(tp=>tp+"--商品--"+p)
    }).foreach(println)

   //  使用join算子
    val res = ordersRDD.join(goodsRDD)
    res.mapPartitionsWithIndex((p,iters)=>{
      iters.map(tp=>tp +"===join后的分区==="+p)
    }).foreach(println)

    println("===========================优化===========================")
     // 1 本地数据
    val mp: collection.Map[Int, String] = goodsRDD.collectAsMap()
    // 遍历订单   拼接商品信息
    val resx = ordersRDD.map(tp => {

      // 每个task中执行
      val order_pid = tp._1
      val order_line = tp._2
      // mp  闭包数据  在每个Task中有一份完整的闭包数据
      val name_des = mp.getOrElse(order_pid, "")
      order_line + "----" + name_des
    })
    resx.foreach(println)
    println("===========================优化===========================")
    // 本地mp集合数据
    // 将mp集合广播出去
    val bs = sc.broadcast(mp)  // main 本地模式   在driver端广播
    val resy = ordersRDD.map(tp => {

      // 每个task中执行
      val order_pid = tp._1
      val order_line = tp._2
      // 获取广播数据
      val mp: collection.Map[Int, String] = bs.value
      val name_des = mp.getOrElse(order_pid, "")
      order_line + "----" + name_des
    })
    resy.foreach(println)


  }

}
